const { Readable, Writable } = require('node:stream');
const fs = require('node:fs').promises;
const path = require('node:path');

class SimpleReadable extends Readable {
  constructor(filename) {
    super();
    this.filename = filename;
    this.fileHandle = null;
    this.position = 0;
    this.init();
  }

  async init() {
    try {
      this.fileHandle = await fs.open(this.filename, 'r');
      console.log('读取文件已打开');
    } catch (err) {
      this.destroy(err);
    }
  }

  async _read(size) {
    try {
      if (!this.fileHandle) {
        // 如果文件还未打开，等待100ms后重试
        setTimeout(() => this._read(size), 100);
        return;
      }

      const buffer = Buffer.alloc(size || 64 * 1024); // 64KB 缓冲区
      const { bytesRead } = await this.fileHandle.read(buffer, 0, buffer.length, this.position);

      if (bytesRead === 0) {
        await this.fileHandle.close();
        this.push(null);
        return;
      }

      this.position += bytesRead;
      this.push(buffer.slice(0, bytesRead));
    } catch (err) {
      this.destroy(err);
    }
  }
}

class SimpleWritable extends Writable {
  constructor(filename) {
    super();
    this.filename = filename;
    this.fileHandle = null;
    this.init();
  }

  async init() {
    try {
      this.fileHandle = await fs.open(this.filename, 'w');
      console.log('写入文件已打开');
    } catch (err) {
      this.destroy(err);
    }
  }

  async _write(chunk, encoding, callback) {
    try {
      if (!this.fileHandle) {
        // 如果文件还未打开，等待100ms后重试
        setTimeout(() => this._write(chunk, encoding, callback), 100);
        return;
      }

      await this.fileHandle.write(chunk);
      callback();
    } catch (err) {
      callback(err);
    }
  }

  _final(callback) {
    const finalizeWrite = async () => {
      try {
        if (this.fileHandle) {
          await this.fileHandle.close();
          console.log('文件已关闭');
        }
        callback();
      } catch (err) {
        callback(err);
      }
    };

    finalizeWrite();
  }
}

// 使用示例
async function transferFile(srcPath, destPath) {
  console.log(`开始传输文件: ${srcPath} -> ${destPath}`);

  const reader = new SimpleReadable(srcPath);
  const writer = new SimpleWritable(destPath);

  reader.on('error', err => console.error('读取错误:', err));
  writer.on('error', err => console.error('写入错误:', err));
  writer.on('finish', () => console.log('文件传输完成'));

  // 数据传输进度显示
  let totalBytes = 0;
  reader.on('data', chunk => {
    totalBytes += chunk.length;
    console.log(`已传输: ${totalBytes} 字节`);
  });

  reader.pipe(writer);
}

// 测试代码
const srcFile = path.join(__dirname, 'input.txt');
const destFile = path.join(__dirname, 'output.txt');

transferFile(srcFile, destFile).catch(err => {
  console.error('传输过程出错:', err);
});